package groupOne.app.DWS;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import groupOne.app.BaseAppOneStream;
import groupOne.bean.homeCourseCount;
import groupOne.common.Constant;
import groupOne.util.AtguiguUtil;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisSink;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Properties;

/**
 * @Author: LixuanFeng
 * @Date: 2022/9/1 17:19
 * @From: GoodGoodStudyDayDayUp
 *
 * 需求 ：统计出 首页浏览人数，商品详情页浏览人数
 *
 *  过滤数据：
 * 	仅保留 page_id 为 home 或 good_detail 的数据
 *
 *  按照 mid 分组
 *
 *  状态中维护每个 mid 维护首页和商品详情页末次访问日期
 *      如果 page_id 为 home，状态中的日期 与 当天日期不相等时 ，homeUvCt =1
 *      如果page_id 为 course_detail, 状态中的日期 与 当天日期不相等时 ，courseUvCt =1
 *  封装到POJO类
 *
 *  添加水印，开窗聚合
 *
 *                   doris 建表语句：
 *                               drop table if exists dws_user_order_unique_window;
 *                               create table if not exists dws_user_order_unique_window
 *                               (
 *                                   `stt`                            DATETIME comment '窗口起始时间',
 *                                   `edt`                            DATETIME comment '窗口结束时间',
 *                                   `cur_date`                       DATE comment '当天日期',
 *                                   `home_uv_ct`                       BIGINT replace comment '首页独立访客数',
 *                                   `course_detail_uv_ct`                BIGINT replace comment '课程详情页独立访客数'
 *                               ) engine = olap aggregate key (`stt`, `edt`, `cur_date`)
 *                               comment "用户主题页面访客汇总表"
 *                               partition by range(`cur_date`)()
 *                               distributed by hash(`stt`) buckets 10 properties (
 *                                 "replication_num" = "3",
 *                                 "dynamic_partition.enable" = "true",
 *                                 "dynamic_partition.time_unit" = "DAY",
 *                                 "dynamic_partition.start" = "-1",
 *                                 "dynamic_partition.end" = "1",
 *                                 "dynamic_partition.prefix" = "par",
 *                                 "dynamic_partition.buckets" = "10",
 *                                 "dynamic_partition.hot_partition_num" = "1"
 *                               );
 */
public class DwsUserPageHomeGodetailWindow_lixuanfeng extends BaseAppOneStream {
    public static void main(String[] args) {
        new DwsUserPageHomeGodetailWindow_lixuanfeng().init(3102,
                2,
                "DwsUserPageHomeGodetailWindow_lixuanfeng",
                Constant.TOPIC_DWD_TRAFFIC_PAGE);
    }


    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        //1.过滤数据：转换成JSON对象，过滤出 home  和 course_detail 的数据
        SingleOutputStreamOperator<JSONObject> filterStream = parseAndFilter(stream);

        //2.分组统计，再把数据封装成POJO类
        SingleOutputStreamOperator<homeCourseCount> pojoStream = countAndPojo(filterStream);

        //3.加水印开窗聚合
        SingleOutputStreamOperator<homeCourseCount> aggStream = WindowAgg(pojoStream);

        //4.写入Doris 没测试
        writeToDoris(aggStream);


    }

    private void writeToDoris(SingleOutputStreamOperator<homeCourseCount> aggStream) {

        Properties props = new Properties();
        props.setProperty("format", "json");
        props.setProperty("strip_outer_array", "true");
        String tableName= "edu.dws_user_page_view_window";

        aggStream
                .map(pojo -> {
                    SerializeConfig config = new SerializeConfig();
                    config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;  // 转成json的时候, 属性名使用下划线
                    return JSON.toJSONString(pojo, config);
                })
                .addSink(DorisSink.sink(new DorisExecutionOptions.Builder()
                                .setBatchIntervalMs(2000L)
                                .setBatchSize(1024 * 1024)
                                .setEnableDelete(false)
                                .setMaxRetries(3)
                                .setStreamLoadProp(props)
                                .build(),
                        new DorisOptions.Builder()
                                .setFenodes(Constant.DORIS_HOST)
                                .setUsername("root")
                                .setPassword("aaaaaa")
                                .setTableIdentifier(tableName)
                                .build()));
        // 写入没测试

    }

    private SingleOutputStreamOperator<homeCourseCount> WindowAgg(SingleOutputStreamOperator<homeCourseCount> pojoStream) {
       return
        pojoStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<homeCourseCount>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((pojo,ts)->pojo.getTs())
                )
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(5L)))
                .reduce(new ReduceFunction<homeCourseCount>() {
                            @Override
                            public homeCourseCount reduce(homeCourseCount value1, homeCourseCount value2) throws Exception {
                                value1.setHomeUvCt(value1.getHomeUvCt() + value2.getHomeUvCt());
                                value2.setCourseDetailUvCt(value1.getCourseDetailUvCt() + value2.getCourseDetailUvCt());
                                return value1;
                            }
                        }
                        ,
                        new ProcessAllWindowFunction<homeCourseCount, homeCourseCount, TimeWindow>() {
                            @Override
                            public void process(Context context, Iterable<homeCourseCount> elements, Collector<homeCourseCount> out) throws Exception {

                                homeCourseCount pojo = elements.iterator().next();

                                String start = AtguiguUtil.toDate(context.window().getStart());
                                String end = AtguiguUtil.toDate(context.window().getEnd());

                                String curdate = AtguiguUtil.toDate(System.currentTimeMillis());

                                pojo.setStt(start);
                                pojo.setEdt(end);
                                pojo.setCurDate(curdate);

                                out.collect(pojo);
                            }
                        }

                );


    }

    private SingleOutputStreamOperator<homeCourseCount> countAndPojo(SingleOutputStreamOperator<JSONObject> filterStream) {

      return
        filterStream
                .keyBy(json->json.getJSONObject("common").getString("mid"))
                .process(new KeyedProcessFunction<String, JSONObject, homeCourseCount>() {

                    private ValueState<String> courseViewDate;
                    private ValueState<String> homeViewDate;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        homeViewDate = getRuntimeContext().getState(new ValueStateDescriptor<String>("homeViewDate", String.class));
                        courseViewDate = getRuntimeContext().getState(new ValueStateDescriptor<String>("courseViewDate", String.class));
                    }

                    @Override
                    public void processElement(JSONObject value, Context ctx, Collector<homeCourseCount> out) throws Exception {

                        Long homeUvCt=0L;

                        Long courseDetailUvCt=0L;

                        String homeLastDate = homeViewDate.value();
                        String courseLastDate = courseViewDate.value();
                        Long todayTS = value.getLong("ts");
                        String today = AtguiguUtil.toDate(todayTS);

                        String pageID = value.getJSONObject("page").getString("page_id");

                        if (!today.equals(homeLastDate) && "home".equals(pageID)){
                            homeUvCt=1L;
                            homeViewDate.update(today);
                        }else if ( !today.equals(courseLastDate) && "course_detail".equals(pageID)){
                            courseDetailUvCt=1L;
                            courseViewDate.update(today);
                        }

                        if (homeUvCt+courseDetailUvCt >=1){
                            out.collect(homeCourseCount.builder()
                                    .courseDetailUvCt(courseDetailUvCt)
                                    .homeUvCt(homeUvCt)
                                    .ts(todayTS)
                                    .build());
                        }
                    }
                });

    }

    private SingleOutputStreamOperator<JSONObject> parseAndFilter(DataStreamSource<String> stream) {

        return
        stream
                .map(jsonStr-> JSON.parseObject(jsonStr))
                .filter(new FilterFunction<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        String pageID = value.getJSONObject("page").getString("page_id");

                        return "home".equals(pageID) || "course_detail".equals(pageID);
                    }
                });

    }
}
